package com.pnnd.stream;

import java.io.IOException;
import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * 线程安全的outputStream
 */
public class NdArrayOutputStream extends MultiplySegmentOutputStream {
    public NdArrayOutputStream(List<MemorySegment> segments) {
        super(segments,0);

        this.operators = new ConcurrentLinkedDeque<>();
    }

    public NdArrayOutputStream(List<MemorySegment> segments,int offset,int limit) {
        super(segments,offset);

        this.operators = new ConcurrentLinkedDeque<>();

        this.limit = limit;
    }

    private Queue<Operator> operators;

    private Thread operatorThread;

    public void addOperator(Operator o){
        this.operators.add(o);

        this.invokeOperator();
    }

    public void removeOperator(Operator o){
        if(this.operators.contains(o)){
            this.operators.remove(o);
        }else{
            throw new RuntimeException("operator has running");
        }
    }

    private void invokeOperator(){
        if(this.operatorThread == null){
            this.operatorThread = new Thread(new Runnable() {
                @Override
                public void run() {

                    Operator operator = NdArrayOutputStream.this.operators.poll();

                    while(operator!=null){
                        if(operator instanceof WriteOperator){
                            WriteOperator writeOperator = (WriteOperator)operator;

                            try{
                                NdArrayOutputStream.super.write(writeOperator.getBytes(),writeOperator.getOffset(),writeOperator.getLength());

                                writeOperator.callback.callback(writeOperator.getSource(),NdArrayOutputStream.this);

                                writeOperator.setResult(new OperatorResult(true));
                            }catch (IOException e){
                                writeOperator.setResult(new OperatorResult(e,false));
                            }
                        }

                        if(operator instanceof FlushOperator){
                            FlushOperator flushOperator = (FlushOperator)operator;

                            try{
                                NdArrayOutputStream.super.flush();

                                flushOperator.getCallBack().callback(flushOperator.getSource(),NdArrayOutputStream.this);

                                flushOperator.setResult(new OperatorResult(true));
                            }catch (IOException e){
                                flushOperator.setResult(new OperatorResult(e,false));
                            }
                        }

                        if(operator instanceof CloseOperator){
                            CloseOperator closeOperator = (CloseOperator)operator;

                            try{
                                NdArrayOutputStream.super.close();

                                closeOperator.getCallBack().callback(closeOperator.getSource(),NdArrayOutputStream.this);

                                closeOperator.setResult(new OperatorResult(true));
                            }catch (IOException e){
                                closeOperator.setResult(new OperatorResult(e,false));
                            }
                        }

                        operator = NdArrayOutputStream.this.operators.poll();
                    }


                    NdArrayOutputStream.this.operatorThread = null;
                }
            });
        }
    }

    public static abstract class Operator{

        private String id;
        private NdArrayOutputStream ndArrayOutputStream;

        public Operator(){
            this.id = UUID.randomUUID().toString();
        }

        @Override
        public int hashCode(){
            return this.id.hashCode();
        }

        @Override
        public boolean equals(Object other){
            if(other instanceof Operator){
                return ((Operator) other).id.equals(this.id);
            }

            return false;
        }

        public abstract OperatorResult getResult();

        public void cancel(){
            this.ndArrayOutputStream.removeOperator(this);
        }
    }

    public static class OperatorResult{
        private Exception exception;
        private boolean success;

        public OperatorResult(Exception e,boolean success){
            this.exception = e;
            this.success = success;
        }

        public OperatorResult(boolean success){
            this.success = success;
        }

        public Exception getException() {
            return exception;
        }

        public void setException(Exception exception) {
            this.exception = exception;
        }

        public boolean isSuccess() {
            return success;
        }

        public void setSuccess(boolean success) {
            this.success = success;
        }
    }

    public static class ObjectOperator<T> extends Operator {

        private T b;
        private Object source;
        private OutputCallback callback;
        private OperatorResult result;


        public T getB() {
            return b;
        }

        public void setB(T b) {
            this.b = b;
        }

        public Object getSource() {
            return source;
        }

        public void setSource(Object source) {
            this.source = source;
        }

        public OutputCallback getCallback() {
            return callback;
        }

        public void setCallback(OutputCallback callback) {
            this.callback = callback;
        }

        public OperatorResult getResult() {
            return result;
        }

        public void setResult(OperatorResult result) {
            this.result = result;
        }
    }

    public static class WriteOperator extends Operator {
        private int offset;
        private int length;
        private byte[] bytes;
        private Object source;
        private OutputCallback callback;
        private OperatorResult result;

        public byte[] getBytes() {
            return bytes;
        }

        public void setBytes(byte[] bytes) {
            this.bytes = bytes;
        }

        public int getOffset() {
            return offset;
        }

        public void setOffset(int offset) {
            this.offset = offset;
        }

        public int getLength() {
            return length;
        }

        public void setLength(int length) {
            this.length = length;
        }

        public Object getSource() {
            return source;
        }

        public void setSource(Object source) {
            this.source = source;
        }

        public OutputCallback getCallback() {
            return callback;
        }

        public void setCallback(OutputCallback callback) {
            this.callback = callback;
        }

        public OperatorResult getResult() {
            return result;
        }

        public void setResult(OperatorResult result) {
            this.result = result;
        }
    }

    public static class FlushOperator extends Operator{
        private Object source;
        private OutputCallback callBack;
        private OperatorResult result;

        public Object getSource() {
            return source;
        }

        public void setSource(Object source) {
            this.source = source;
        }

        public OutputCallback getCallBack() {
            return callBack;
        }

        public void setCallBack(OutputCallback callBack) {
            this.callBack = callBack;
        }

        public OperatorResult getResult() {
            return result;
        }

        public void setResult(OperatorResult result) {
            this.result = result;
        }
    }

    public static class CloseOperator extends Operator{
        private Object source;
        private OutputCallback callBack;
        private OperatorResult result;

        public Object getSource() {
            return source;
        }

        public void setSource(Object source) {
            this.source = source;
        }

        public OutputCallback getCallBack() {
            return callBack;
        }

        public void setCallBack(OutputCallback callBack) {
            this.callBack = callBack;
        }

        public OperatorResult getResult() {
            return result;
        }

        public void setResult(OperatorResult result) {
            this.result = result;
        }
    }

    class OperatorFuture implements Future<OperatorResult> {

        private Operator operator;
        private boolean canceled;

        public OperatorFuture(Operator operator){
            this.operator = operator;
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            try {
                this.operator.cancel();
                this.canceled = true;
                return true;
            }catch (RuntimeException e){
                return false;
            }
        }

        @Override
        public boolean isCancelled() {
            return this.canceled;
        }

        @Override
        public boolean isDone() {
            return this.operator.getResult() != null;
        }

        @Override
        public OperatorResult get() throws InterruptedException, ExecutionException {

            return this.operator.getResult();
        }

        @Override
        public OperatorResult get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {

            this.await(timeout,unit);

            return this.get();
        }

        public void await(long timeout, TimeUnit unit){
            if(timeout == -1){
                while (isDone());

                return;
            }

            long currentTime = System.currentTimeMillis();

            while((System.currentTimeMillis()-currentTime)>0 || !isDone());

            return;
        }
    }

    public Future<OperatorResult> asyncWrite(int b,OutputCallback callback,Object source){
        byte[] bs = new byte[]{(byte)b};

        WriteOperator writeOperator = new WriteOperator();

        writeOperator.setBytes(bs);
        writeOperator.setOffset(0);
        writeOperator.setLength(1);
        writeOperator.setCallback(callback);
        writeOperator.setSource(source);

        OperatorFuture future = new OperatorFuture(writeOperator);

        this.addOperator(writeOperator);

        return future;
    }

    public Future<OperatorResult> asyncWrite(int b){
        return this.asyncWrite(b,null,this);
    }

    public Future<OperatorResult> asyncWrite(byte[] bs){
        return this.asyncWrite(bs,null,this);
    }

    public Future<OperatorResult> asyncWrite(byte[] bs,OutputCallback callback,Object source){
        return this.asyncWrite(bs,0,bs.length,callback,source);
    }

    public Future<OperatorResult> asyncWrite(byte[] bs,int offset,int len){
        return this.asyncWrite(bs,offset,len,null,this);
    }

    public Future<OperatorResult> asyncWrite(byte[] bs,int offset,int len,OutputCallback callback,Object source){
        WriteOperator writeOperator = new WriteOperator();

        writeOperator.setBytes(bs);
        writeOperator.setOffset(offset);
        writeOperator.setLength(len);
        writeOperator.setCallback(callback);
        writeOperator.setSource(source);

        OperatorFuture future = new OperatorFuture(writeOperator);

        this.addOperator(writeOperator);

        return future;
    }

    public Future<OperatorResult> asyncFlush(OutputCallback callback,Object source){
        FlushOperator flushOperator = new FlushOperator();

        flushOperator.setCallBack(callback);

        flushOperator.setSource(source);

        OperatorFuture future = new OperatorFuture(flushOperator);

        this.addOperator(flushOperator);

        return future;
    }

    public Future<OperatorResult> asyncFlush(){
        return this.asyncFlush(null,this);
    }

    public Future<OperatorResult> asyncClose(){
        return this.asyncClose(null,this);
    }

    public Future<OperatorResult> asyncClose(OutputCallback callback,Object source){
        CloseOperator closeOperator = new CloseOperator();

        closeOperator.setCallBack(callback);

        closeOperator.setSource(source);

        OperatorFuture future = new OperatorFuture(closeOperator);

        this.addOperator(closeOperator);

        return future;
    }

    @Override
    public void close(){
        Future<OperatorResult> result = this.asyncClose();

        try {
            result.get(-1,TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void flush(){
        Future<OperatorResult> result = this.asyncFlush();

        try {
            result.get(-1,TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(int value){
        Future<OperatorResult> result = this.asyncWrite(value);

        try {
            result.get(-1,TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(byte[] value){
        Future<OperatorResult> result = this.asyncWrite(value);

        try {
            result.get(-1,TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(byte[] value,int offset,int len){
        Future<OperatorResult> result = this.asyncWrite(value,offset,len);

        try {
            result.get(-1,TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
